package util

import (
	"common/goredis"
	"common/logging"

	"errors"
	"fmt"
	"github.com/Shopify/sarama"
	"github.com/golang/protobuf/proto"
	"iot/iot_app_server/proto"
	"strings"
)

var (
	route_redis_handle *goredis.Redis
	route_producer     sarama.SyncProducer
)

func InitRoute(BrokerAddr []string, redisurl string) {
	InitRouteProducer(BrokerAddr)
	InitRouteRedis(redisurl)
}

func InitRouteProducer(BrokerAddr []string) bool {
	if len(BrokerAddr) == 0 {
		return false
	}
	var err error
	route_producer, err = sarama.NewSyncProducer(BrokerAddr, nil)
	if err != nil {
		logging.Error("InitProducer failed")
		return false
	}

	return true
}

func CloseProducer() {
	if route_producer != nil {
		route_producer.Close()
	}
}

func PubMsg(topic, data string) error {
	logging.Info("PubMsg topic=%s", topic)
	msg := &sarama.ProducerMessage{Topic: topic, Value: sarama.StringEncoder(data)}
	_, _, err := route_producer.SendMessage(msg)
	if err != nil {
		logging.Error("FAILED to send message,topic=%s,data=%s,err=%s\n", topic, data, err.Error())
		return err
	}
	return nil
}

func InitRouteRedis(redisurl string) {
	var err error
	route_redis_handle, err = goredis.DialURL(redisurl)
	if err != nil {
		logging.Error("redis init fail:%s,%s", redisurl, err.Error())
		panic("InitRedis failed")
		return
	} else {
		logging.Info("redis conn ok:%s", redisurl)
	}

}

func RouteToDevice(reqData *MSGBUS_DOWNSTREAM.RouteMsgInfo_B2B_C) (Code int32, RepMsg string) {
	Code = 0
	RepMsg = "ok"
	deviceId := reqData.DeviceId
	sessionid_b := reqData.SessionId
	var sessionid string
	if sessionid_b == nil || string(sessionid_b) == "" {
		var err_s error
		sessionid, err_s = GetDeviceSessionid(string(deviceId))
		if err_s != nil {
			Code = 1007
			RepMsg = "get sessionid failed"
			return Code, RepMsg
		}
	} else {
		sessionid = string(sessionid_b)
	}

	//设备不在线，不处理
	if sessionid == "" {
		RepMsg = "device offline"
		return Code, RepMsg
	}
	adapter, err := GetAdapterBySessionid(sessionid)
	if err != nil || adapter == "" {
		Code = 1008
		RepMsg = "get adapter failed"
		return Code, RepMsg
	}

	rawData := &MSGBUS_DOWNSTREAM.MsgB2BDownstreamData{
		B2BTopic:   reqData.Topic,
		B2BPayload: reqData.Payload,
	}
	rawDataBytes, _ := proto.Marshal(rawData)

	sendData := &MSGBUS_DOWNSTREAM.B2BSendMsg{
		Controltype: MSGBUS_DOWNSTREAM.CONTROL_TYPE_control_type_data.Enum(),
		Sendtype:    reqData.IsStore,
		Deviceid:    reqData.DeviceId,
		Sessionid:   proto.String(sessionid),
		Body:        rawDataBytes,
	}
	sendDataBytes, _ := proto.Marshal(sendData)

	go PubMsg(adapter, string(sendDataBytes))
	return Code, RepMsg
}

func DeviceidToString(deviceid []byte) string {
	length := len(deviceid)
	if length == 0 {
		return ""
	}
	ret_b := make([]byte, length)
	for i := 0; i < length; i++ {
		ret_b[i] = deviceid[length-1-i]
	}
	ret_s := fmt.Sprintf("%0X", ret_b)

	return ret_s
}

//返回 空字符串，并且error为nil表示redis中没有session信息，即设备不在线
func GetDeviceSessionid(deviceid string) (string, error) {
	jsonSessionInfo, err := route_redis_handle.Get("devid2session_" + deviceid)
	if err != nil {
		logging.Error("GetDeviceSessionid redis_handle.get failed,deviceid=%s, err: %s", deviceid, err.Error())
		return "", err
	}
	logging.Debug("GetDeviceSessionid deviceid=%s,jsonSessionInfo=%s", string(deviceid), string(jsonSessionInfo))

	//设备不在线
	if string(jsonSessionInfo) == "" {
		return "", nil
	}

	return string(jsonSessionInfo), nil
}

//session格式：1234@adapter.com/4321/abc
func GetAdapterBySessionid(sessionid string) (string, error) {
	logging.Debug("GetAdapterBySessionid sessionid=%s", sessionid)
	atPos := strings.Index(sessionid, "@")
	if atPos == -1 {
		logging.Error("GetAdapterBySessionid invalid sessonid:%s", sessionid)
		return "", errors.New("invalid sessonid")
	}

	specPos := strings.Index(sessionid, "/")
	if specPos == -1 {
		logging.Error("GetAdapterBySessionid invalid sessonid:%s", sessionid)
		return "", errors.New("invalid sessonid")
	}

	if atPos+1 >= specPos {
		logging.Error("GetAdapterBySessionid invalid sessonid:%s", sessionid)
		return "", errors.New("invalid sessonid")
	}

	return sessionid[atPos+1 : specPos], nil
}
